package com.chif.demo.producer;

import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.exception.RemotingException;

import java.util.concurrent.TimeUnit;

/**
 * 生产者异步发送消息
 */
public class AsyncProducer {

    public static void main(String[] args) throws Exception {
        //创建一个producer，参数为Producer Group名称
        DefaultMQProducer producer = new DefaultMQProducer("pg");

        producer.setNamesrvAddr("192.168.10.100:9876");

        //指定异步发送失败后不进行重试发送
        producer.setRetryTimesWhenSendAsyncFailed(0);

        //指定新创建topic的Queue数量为2
        producer.setDefaultTopicQueueNums(2);

        //开启生产者
        producer.start();

        for (int i = 0; i < 100; i++) {
            byte[] body = ("Hi," + i).getBytes();
            Message msg = new Message("myTopicA", "myTag", body);

            //异步发送消息
            try {
                producer.send(msg, new SendCallback() {
                    //当producer 接收到mq发送来的ack后就会触发回调方法的执行
                    public void onSuccess(SendResult sendResult) {
                        System.out.println(sendResult);
                    }

                    public void onException(Throwable e) {
                        e.printStackTrace();
                    }
                });
            } catch (MQClientException e) {
                e.printStackTrace();
            } catch (RemotingException e) {
                e.printStackTrace();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }

        TimeUnit.SECONDS.sleep(3);
        //关闭producer
        producer.shutdown();

    }
}
